package com.demo.expertsystemdemo.mqtt.config;

import com.demo.expertsystemdemo.mqtt.entity.Config;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * @author lzb22
 */
@Configuration
@Slf4j
@EnableConfigurationProperties(Config.class)
public class MqttAutoConfiguration {

    /**
     * 入站通道名（消费者）
     */
    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";
    /**
     * 出站通道名（生产者）
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
    public static final String FACTORY_NAME = "mqttPahoClientFactory";
    public static final String OPTIONS_NAME = "mqttConnectOptions";

    private final Config config;

    public MqttAutoConfiguration(Config config) {
        this.config = config;
    }

    /**
     * MQTT连接器选项
     */
    @Bean(name = OPTIONS_NAME)
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(config.getUsername());
        options.setPassword(config.getPassword().toCharArray());
        options.setServerURIs(config.getHost());
        options.setConnectionTimeout(config.getTimeout());
        options.setKeepAliveInterval(config.getKepAliveInterval());
        options.setAutomaticReconnect(config.getAutomaticReconnect() == null || config.getAutomaticReconnect());
        return options;
    }

    /**
     * MQTT客户端
     */
    @Bean(name = FACTORY_NAME)
    public MqttPahoClientFactory mqttPahoClientFactory(
            @Qualifier(OPTIONS_NAME) MqttConnectOptions options) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * =====================================================
     * MQTT消费端订阅通道（消息入站）
     * =====================================================
     */
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel inboundChannel() {
        return new DirectChannel();
    }

    /**
     * =====================================================
     * MQTT消费端连接配置【消息入站】
     * =====================================================
     */
    @Bean
    public MessageProducer inbound(
            @Qualifier(CHANNEL_NAME_IN) MessageChannel channel,
            @Qualifier(FACTORY_NAME) MqttPahoClientFactory factory) {
        // 可以同时消费（订阅）多个Topic
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(config.getConsumerClientId()
                  + "_in_bound", factory, config.getTopics());
        // 设置操作完成的超时时长，默认30000毫秒
        adapter.setCompletionTimeout(3000);
        // 配置默认消息转换器(qos=0, retain=false, charset=UTF-8)
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        // 设置订阅通道
        adapter.setOutputChannel(channel);
        return adapter;
    }


    /**
     * MQTT生产端发布通道（消息出站）
     */
    @Bean(name = CHANNEL_NAME_OUT)
    public MessageChannel outboundChannel() {
        return new DirectChannel();
    }


    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
    public MessageHandler outbound(
            @Qualifier(FACTORY_NAME) MqttPahoClientFactory factory) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(config.getProducerClientId()
                + "_out_bound", factory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultRetained(false);
        messageHandler.setAsyncEvents(false);

        messageHandler.setDefaultQos(2);
        messageHandler.setDefaultTopic(config.getProducerDefaultTopic());
        return messageHandler;
    }

}
